package com.yifeng.repo.toolkit.mq;

import com.alibaba.cloud.stream.binder.rocketmq.constant.RocketMQConst;
import com.google.common.base.Strings;
import com.yifeng.repo.base.utils.common.Snowflake;
import com.yifeng.repo.toolkit.mq.configure.BindingProperties;
import com.yifeng.repo.toolkit.mq.configure.MqProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Created by daibing on 2022/9/25.
 */
@Slf4j
public class DefaultMessageProducer {
    private static final String RABBIT_BINDER_TYPE = "rabbit";
    private static final String ROUTING_KEY = "x-routing-key";
    private static final Integer DELAY_TIME_LEVEL_MIN = 1;
    private static final Integer DELAY_TIME_LEVEL_MAX = 18;
    private final ConcurrentMap<String, BindingProperties> topic2Binding = new ConcurrentHashMap<>();
    private final MqProperties properties;
    private final StreamBridge streamBridge;
    private final Integer[] delayTime = new Integer[]{ 1, 5, 10, 30, 60, 2*60, 3*60, 4*60, 5*60, 6*60, 7*60, 8*60, 9*60, 10*60, 20*60, 30*60, 60*60, 2*60*60};

    public DefaultMessageProducer(MqProperties properties, List<BindingProperties> producerBindingPropertiesList, StreamBridge streamBridge) {
        for (BindingProperties bindingProperties : producerBindingPropertiesList) {
            topic2Binding.put(bindingProperties.getDestination(), bindingProperties);
        }
        this.properties = properties;
        this.streamBridge = streamBridge;
    }

    /**
     * 发送普通消息
     * @param topic 消息主题
     * @param tag 消息标签用在 Consumer的场景中，用于服务端消息过滤。（补充：配置发送顺序消息后该方法默认使用tag作为路由键）
     * @param key 消息在业务层面的唯一标识，业务可以根据这个Key查找消息，要求Key唯一。注意RocketMQ不保证message id唯一
     * @param body 消息体
     * @param headers 消息头
     */
    public Long sendMsg(String topic, String tag, String key, String body, Map<String, Object> headers) {
        return this.send(topic, tag, key, null, body, 0, null, null, headers);
    }

    /**
     * 发送顺序消息
     * @param routingKey 路由键，没有指定路由键的场景下默认使用tag作为路由键
     */
    public Long sendOrderlyMsg(String topic, String tag, String key, String routingKey, String body, Map<String, Object> headers) {
        return this.send(topic, tag, key, routingKey, body, 0, null, null, headers);
    }

    /**
     * 发送延时消息
     * @param delayTimeLevel 1 ~ 18，对应的值1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public Long sendDelayMsg(String topic, String tag, String key, String body, int delayTimeLevel, Map<String, Object> headers) {
        if (delayTimeLevel < DELAY_TIME_LEVEL_MIN || delayTimeLevel > DELAY_TIME_LEVEL_MAX) {
            throw new IllegalArgumentException("Error delayTimeLevel=" + delayTimeLevel + " by tag=" + tag);
        }
        return this.send(topic, tag, key, null, body, delayTimeLevel, null, null, headers);
    }

    /**
     * 发送事务消息
     * @param userTransactionalArgs 事务场景参数，标识一个业务场景
     * @param transactionId 事务id
     */
    public Long sendTxMsg(String topic, String tag, String key, String body, String userTransactionalArgs, String transactionId, Map<String, Object> headers) {
        return this.send(topic, tag, key, null, body, 0, userTransactionalArgs, transactionId, headers);
    }

    /**
     * 发送消息
     */
    protected Long send(String topic, String tag, String key,
                        String routingKey, String body, int delayTimeLevel,
                        String userTransactionalArgs, String transactionId,
                        Map<String, Object> headers) {
        // match bindingName
        BindingProperties bindingProperties = topic2Binding.get(topic);
        if (bindingProperties == null) {
            throw new IllegalArgumentException("Not found binding by topic=" + topic);
        }

        // build message
        Message<String> msg = null;
        Long msgId = Snowflake.get().nextId();
        MimeType mimeType = this.getMimeType(bindingProperties.getContentType());
        if (RABBIT_BINDER_TYPE.equalsIgnoreCase(bindingProperties.getBinderType())) {
            msg = this.buildRabbitMessage(msgId, tag, key, routingKey, body, delayTimeLevel, userTransactionalArgs, transactionId, headers, mimeType);
        } else {
            msg = this.buildRocketMessage(msgId, tag, key, routingKey, body, delayTimeLevel, userTransactionalArgs, transactionId, headers, mimeType);
        }

        // send message
        streamBridge.send(bindingProperties.getBindingName(), msg);
        return msgId;
    }

    protected Message<String> buildRocketMessage(Long msgId, String tag, String key,
                                                 String routingKey, String body, int delayTimeLevel,
                                                 String userTransactionalArgs, String transactionId,
                                                 Map<String, Object> headers, MimeType mimeType) {
        MessageBuilder<String> builder = MessageBuilder.withPayload(body)
                .setHeader(MessageHeaders.CONTENT_TYPE, mimeType)
                .setHeader(MessageConst.PROPERTY_TAGS, tag)
                .setHeader(MessageConst.PROPERTY_KEYS, key)
                .setHeader(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, msgId);
        if (!Strings.isNullOrEmpty(routingKey)) {
            builder.setHeader(ROUTING_KEY, routingKey);
        }
        if (delayTimeLevel >= DELAY_TIME_LEVEL_MIN && delayTimeLevel <= DELAY_TIME_LEVEL_MAX) {
            builder.setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel);
        }
        if (!Strings.isNullOrEmpty(userTransactionalArgs) && !Strings.isNullOrEmpty(transactionId)) {
            builder.setHeader(RocketMQConst.USER_TRANSACTIONAL_ARGS, userTransactionalArgs);
            builder.setHeader(RocketMQConst.Headers.TRANSACTION_ID, transactionId);
        }
        for (Map.Entry<String, Object> entry : headers.entrySet()) {
            builder.setHeader(properties.getBizHeaderPrefix() + entry.getKey(), entry.getValue());
        }
        return builder.build();
    }

    protected Message<String> buildRabbitMessage(Long msgId, String tag, String key,
                                                 String routingKey, String body, int delayTimeLevel,
                                                 String userTransactionalArgs, String transactionId,
                                                 Map<String, Object> headers, MimeType mimeType) {
        MessageBuilder<String> builder = MessageBuilder.withPayload(body)
                .setHeader(MessageHeaders.CONTENT_TYPE, mimeType)
                .setHeader("RABBIT_TAGS", tag)
                .setHeader("RABBIT_KEYS", key);
        if (!Strings.isNullOrEmpty(routingKey)) {
            builder.setHeader(ROUTING_KEY, routingKey);
        }
        if (delayTimeLevel >= DELAY_TIME_LEVEL_MIN && delayTimeLevel <= DELAY_TIME_LEVEL_MAX) {
            builder.setHeader("x-delay", delayTime[delayTimeLevel - 1]);
        }
        for (Map.Entry<String, Object> entry : headers.entrySet()) {
            builder.setHeader(properties.getBizHeaderPrefix() + entry.getKey(), entry.getValue());
        }
        return builder.build();
    }

    protected MimeType getMimeType(String contentType) {
        if (MimeTypeUtils.APPLICATION_JSON_VALUE.equalsIgnoreCase(contentType)) {
            return MimeTypeUtils.APPLICATION_JSON;
        } else if (MimeTypeUtils.APPLICATION_XML_VALUE.equalsIgnoreCase(contentType)) {
            return MimeTypeUtils.APPLICATION_XML;
        } else {
            return MimeTypeUtils.APPLICATION_OCTET_STREAM;
        }
    }

}
